package com.itheima.delaytask.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.itheima.cache.CacheService;
import com.itheima.delaytask.constants.DelayTaskRedisKeyConstants;
import com.itheima.delaytask.constants.TaskStatusConstants;
import com.itheima.delaytask.constants.ZKdirConstants;
import com.itheima.delaytask.dto.Task;
import com.itheima.delaytask.exception.ScheduleSystemException;
import com.itheima.delaytask.exception.TaskNotExistException;
import com.itheima.delaytask.mapper.TaskInfoLogsMapper;
import com.itheima.delaytask.mapper.TaskInfoMapper;
import com.itheima.delaytask.po.TaskInfo;
import com.itheima.delaytask.po.TaskInfoLogs;
import com.itheima.delaytask.properties.SystemParamProperties;
import com.itheima.delaytask.service.TaskService;
import com.itheima.utils.ZKSelectMasterUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Created by 传智播客*黑马程序员.
 */
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskInfoMapper taskInfoMapper;

    @Autowired
    private TaskInfoLogsMapper taskInfoLogsMapper;

    @Autowired
    private CacheService cacheService;

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Autowired
    private SystemParamProperties paramProperties;

    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    private long nextScheduleTime;//下一次未来2分钟的时间节点

    @Autowired
    private ZKSelectMasterUtil zkSelectMasterUtil;

    @PostConstruct
    public void syncData() {
        //开始选主
        zkSelectMasterUtil.selectMaster(ZKdirConstants.schedule_leaderPath);
        //注册也需要有一定时间,为了后面的操作不会因为这个时间差造成误判,在这个地方宁愿等上个1两秒
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
        }

        //首次运行即执行一次，然后按照设定的时间周期按照固定频率执行
        threadPoolTaskScheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (zkSelectMasterUtil.checkMaster(ZKdirConstants.schedule_leaderPath)) {
                    log.info("主节点{}开始进行定时数据恢复", LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
                    reloadData();
                } else {
                    log.info("从节点备份");
                }
            }
        }, TimeUnit.MINUTES.toMillis(paramProperties.getPreLoad()));
    }

    private void reloadData() {
        long start = System.currentTimeMillis();
        //将数据库中的任务数据恢复到缓存中
        //清理缓存中原有的数据
        clearCache();

        //查询数据库中的所有分组信息
        QueryWrapper<TaskInfo> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda()
                .select(TaskInfo::getTaskType, TaskInfo::getPriority)
                .groupBy(TaskInfo::getTaskType, TaskInfo::getPriority);
        List<TaskInfo> group_task = taskInfoMapper.selectList(queryWrapper);

        if (Objects.isNull(group_task)) {
            return;
        }

        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, paramProperties.getPreLoad());
        nextScheduleTime = calendar.getTimeInMillis();

        //将这个变量存储在redis中
        cacheService.set(DelayTaskRedisKeyConstants.nextScheduleTime, nextScheduleTime + "");

        CountDownLatch countDownLatch = new CountDownLatch(group_task.size());
        for (TaskInfo taskInfo : group_task) {
            taskExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    //查询该分组下的所有任务数据
                    QueryWrapper<TaskInfo> queryWrapper = new QueryWrapper<>();
                    queryWrapper.lambda()
                            .eq(TaskInfo::getTaskType, taskInfo.getTaskType())
                            .eq(TaskInfo::getPriority, taskInfo.getPriority())
                            .le(TaskInfo::getExecuteTime, calendar.getTime());
                    List<TaskInfo> taskInfos = taskInfoMapper.selectList(queryWrapper);

                    for (TaskInfo info : taskInfos) {
                        Task task = new Task();
                        BeanUtils.copyProperties(info, task);
                        task.setExecuteTime(info.getExecuteTime().getTime());
                        addTaskToCache(task);
                    }

                    countDownLatch.countDown();
                }
            });
        }

        try {
            countDownLatch.await(5, TimeUnit.MINUTES);
            log.info("多线程分组数据恢复耗时:{}", System.currentTimeMillis() - start);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void clearCache() {
        //删除所有的key 
        Set<String> futureKeys = cacheService.scan(DelayTaskRedisKeyConstants.FUTURE + "*");
        Set<String> topicKeys = cacheService.scan(DelayTaskRedisKeyConstants.TOPIC + "*");
        cacheService.delete(futureKeys);
        cacheService.delete(topicKeys);
    }

    @Override
    @Transactional
    public long addTask(Task task) throws ScheduleSystemException {

        boolean success = addTaskToDb(task);
        if (success) {
            addTaskToCache(task);
        }
        return task.getTaskId();
    }

    private void addTaskToCache(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();

        long nextScheduleTime = fetchNextScheduleTime();

        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            //任务添加到消费者队列List  左进右出
            cacheService.lLeftPush(DelayTaskRedisKeyConstants.TOPIC + key, JSON.toJSONString(task));
        } else if (task.getExecuteTime() <= nextScheduleTime) {
            //任务数据添加到zset集合中排序等待消费
            cacheService.zAdd(DelayTaskRedisKeyConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
        }
    }

    private long fetchNextScheduleTime() {
        Boolean exists = cacheService.exists(DelayTaskRedisKeyConstants.nextScheduleTime);
        if (exists) {

            String nextScheduleTime = cacheService.get(DelayTaskRedisKeyConstants.nextScheduleTime);
            log.info("从缓存中获取了nextScheduleTime:{}", nextScheduleTime);
            return Long.parseLong(nextScheduleTime);
        } else {
            //数据的补偿
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, paramProperties.getPreLoad());
            this.nextScheduleTime = calendar.getTimeInMillis();
            cacheService.set(DelayTaskRedisKeyConstants.nextScheduleTime, this.nextScheduleTime + "");
            log.info("缓存中没有nextScheduleTime,数据补偿{}", nextScheduleTime);
            return this.nextScheduleTime;
        }

    }

    private boolean addTaskToDb(Task task) {
        boolean success = false;

        try {
            //将任务数据添加到taskinfo表中
            TaskInfo taskInfo = new TaskInfo();
            taskInfo.setTaskType(task.getTaskType());
            taskInfo.setPriority(task.getPriority());
            taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskInfo.setParameters(task.getParameters());

            taskInfoMapper.insert(taskInfo);

            //设置任务的id
            task.setTaskId(taskInfo.getTaskId());

            //向taskinfo_logs表中保存数据
            TaskInfoLogs taskInfoLogs = new TaskInfoLogs();
            BeanUtils.copyProperties(taskInfo, taskInfoLogs);
            taskInfoLogs.setVersion(1);
            taskInfoLogs.setStatus(TaskStatusConstants.SCHEDULED);

            taskInfoLogsMapper.insert(taskInfoLogs);
            success = true;
        } catch (Exception e) {
            log.error("add task error,task={}", task);
            throw new ScheduleSystemException(e);
        }
        return success;
    }

    @Override
    @Transactional
    public boolean cancelTask(long taskId) throws TaskNotExistException, ScheduleSystemException {
        boolean success = false;
        /**
         * 更改数据库taskinfo_logs表中任务的状态为2,同时要删除taskinfo表中的任务数据
         * 删除缓存中的任务数据
         */
        Task task = updateDb(taskId, TaskStatusConstants.CANCELLED);
        if (task != null) {
            removeFromCache(task);
            success = true;
        }
        return success;
    }

    private void removeFromCache(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            //从消费者队列List移除数据
            cacheService.lRemove(DelayTaskRedisKeyConstants.TOPIC + key, 0, JSON.toJSONString(task));
        } else {
            //从zset集合中移除数据
            cacheService.zRemove(DelayTaskRedisKeyConstants.FUTURE + key, JSON.toJSONString(task));
        }

    }

    private Task updateDb(long taskId, int status) {
        Task task = null;
        //根据任务id查询任务日志
        TaskInfoLogs taskInfoLogs = taskInfoLogsMapper.selectById(taskId);
        if (taskInfoLogs == null) {
            throw new TaskNotExistException("task not exist,taskId={}", taskId);
        }

        try {
            //更改任务日志状态
            UpdateWrapper<TaskInfoLogs> updateWrapper = new UpdateWrapper<>();
            updateWrapper.lambda()
                    .set(TaskInfoLogs::getStatus, status)
                    .eq(TaskInfoLogs::getTaskId, taskInfoLogs.getTaskId());
            taskInfoLogsMapper.update(null, updateWrapper);

            //删除任务表中的数据
            taskInfoMapper.deleteById(taskId);
            task = new Task();
            BeanUtils.copyProperties(taskInfoLogs, task);
            task.setExecuteTime(taskInfoLogs.getExecuteTime().getTime());
        } catch (Exception e) {
            log.error("cancel task error,taskId={}", taskId);
            throw new ScheduleSystemException(e);
        }
        return task;
    }

    @Override
    @Transactional
    public Task poll() throws ScheduleSystemException {
        Task task = null;

        //查询当前要执行的任务数据的集合
        Set<String> byScore = cacheService.zRangeByScore(DelayTaskRedisKeyConstants.DBCACHE, 0, System.currentTimeMillis());
        if (byScore == null || byScore.size() == 0) {
            return null;
        }

        try {
            //获取集合中的第一个元素
            String task_json = byScore.iterator().next();
            task = JSON.parseObject(task_json, Task.class);

            //更改数据库信息
            updateDb(task.getTaskId(), TaskStatusConstants.EXECUTED);
            //从缓存中移除任务数据
            cacheService.zRemove(DelayTaskRedisKeyConstants.DBCACHE, task_json);
        } catch (Exception e) {
            log.error("pull task error,msg={}", e.getMessage());
            throw new ScheduleSystemException(e);
        }
        return task;
    }

    @Override
    @Transactional
    public Task poll(int type, int priority) throws ScheduleSystemException {
        Task task = null;
        try {
            //从消费者队列List中找到对用的分组，从该分组下或当前要执行的任务
            String key = type + "_" + priority;
            String task_json = cacheService.lRightPop(DelayTaskRedisKeyConstants.TOPIC + key);
            if (StringUtils.isEmpty(task_json)) {
                return null;
            }

            task = JSON.parseObject(task_json, Task.class);
            //更改数据库信息,
            updateDb(task.getTaskId(), TaskStatusConstants.EXECUTED);
        } catch (Exception e) {
            log.error("poll task error,type={},priority={}", type, priority);
            throw new ScheduleSystemException(e);
        }
        return task;
    }

    @Override
    //@Scheduled(cron = "*/1 * * * * ?")
    public void refresh() {
        taskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                /**
                 * - 使用scan找到未来数据集合的所有key
                 - 根据每一个key从该分组下 获取当前需要执行的任务数据Set集合
                 - 将这组数据添加到消费者队列的对应分组上，并从zset中移除
                 */

                log.info("{}进行了定时刷新", LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
               /*try {
                   TimeUnit.SECONDS.sleep(2);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }*/
                Set<String> future_keys = cacheService.scan(DelayTaskRedisKeyConstants.FUTURE + "*");//future_*

                if (future_keys == null || future_keys.size() == 0) {
                    return;
                }

                for (String future_key : future_keys) {
                    Set<String> values = cacheService.zRangeByScore(future_key, 0, System.currentTimeMillis());

                    if (values == null || values.size() == 0) {
                        continue;
                    }
                    //future_key future_2000_2    topic_2000_2
                    String topicKey = DelayTaskRedisKeyConstants.TOPIC + future_key.split(DelayTaskRedisKeyConstants.FUTURE)[1];
                    /*for (String value : values) {
                        cacheService.lLeftPush(topicKey,value);
                        cacheService.zRemove(future_key,value);
                    }*/
                    cacheService.refreshWithPipeline(future_key, topicKey, values);
                }

            }
        });

    }

}
